package com.gis507.mapreduce.AISToMySql;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class AISInterpolationDriver {

    //String mmsi, String shipType, String nacStatusCN, double draught,
    // double heading, double course, double speed, String dest,
    // long unixTime, double lon_d, double lat_d, int seaRange
    private static String tableName = "PassengerShip";
    private static String [] fields = {"MMSI" ,"shipType","nacStatusEN","draught","heading","course","speed","dest",
            "unixTime","lon_d","lat_d","seaRange"};

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        //12、通过类得加载器加载配置文件
        InputStream is = AISInterpolationDriver.class.getClassLoader().getResourceAsStream("jdbc.properties");

        //实例化一个Properties对象
        Properties pros = new Properties();

        //将数据流放到pros中
        pros.load(is);

        //获取数据
        String user = pros.getProperty("user");
        String password = pros.getProperty("password");
        String url = pros.getProperty("url");
        String driverClass = pros.getProperty("driverClass");

        Class.forName(driverClass);

        //1、新建Job
        Configuration conf = new Configuration();

        //13、创建mysql连接
        DBConfiguration.configureDB(conf,driverClass,url,user,password);

        //10、key和value之间使用"，"
//        conf.set("mapred.textoutputformat.separator", ",");

        //11、mapreduce的内存设置
        conf.set("mapreduce.map.memory.mb","8192");
        conf.set("mapreduce.reduce.memory.mb","8192");
//        mapred.child.java.opts;
        conf.set("mapred.child.java.opts","-Xmx1024m");

        Job job = Job.getInstance(conf);

        //2、设置jar包
        job.setJarByClass(AISInterpolationDriver.class);

        //3、关联map和reduce关系
        job.setMapperClass(AISInterpolationMapper.class);
        job.setReducerClass(AISInterpolationReducer.class);

        //4、设置map的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(AisClassifyBean.class);

        //5、设置最终的输出类型
        job.setOutputKeyClass(AisClassifyBean.class);
        job.setOutputValueClass(NullWritable.class);

        //6、设置输入输出路径
        FileInputFormat.setInputPaths(job,new Path("F:\\Done\\202002\\part-r-00006"));
        FileOutputFormat.setOutputPath(job,new Path("F:\\Done\\dic"));
        //写入mysql中
//        DBOutputFormat.setOutput(job,tableName,fields);

        //7、提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0:1);
    }
}
